package com.bigdata.warehouse.dwd;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdEntranceGuardLog {
    public static void main(String[] args) {
        //1.获取Stream的执行环境
        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
        senv.setParallelism(1);

        //2.创建表的执行环境
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(senv);

        //开启checkpoint容错
//        senv.enableCheckpointing(60000);
//        senv.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        senv.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);
//        senv.getCheckpointConfig().setCheckpointTimeout(10000);
//        senv.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

        //设置状态后端
        //（1）开启RocksDB
//        senv.setStateBackend(new EmbeddedRocksDBStateBackend());
//        //（2）设置checkpoint存储位置
//        senv.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage("hdfs://mycluster/flink/checkpoints"));

        //3.读取门禁刷卡事实表ods_entrance_guard_log
        tEnv.executeSql("CREATE TABLE ods_entrance_guard_log(\n" +
                " id STRING,\n" +
                " openMode SMALLINT,\n" +
                " openTime STRING,\n" +
                " ownerId INT,\n" +
                " proc_time as PROCTIME()\n" +
                ") WITH (\n" +
                " 'connector' = 'kafka',\n" +
                " 'topic' = 'ods_entrance_guard_log',\n" +
                " 'properties.bootstrap.servers' = 'master:9092',\n" +
                " 'properties.group.id' = 'ods_entrance_guard_log',\n" +
                " 'scan.startup.mode' = 'earliest-offset',\n" +
                " 'format' = 'json'\n" +
                ")\n");
        //4.读取用户维表
        tEnv.executeSql("CREATE TABLE dim_base_ower(\n" +
                " id INT,\n" +
                " roomId INT,\n" +
                " owerCode STRING,\n" +
                " owerName STRING,\n" +
                " owerSex STRING,\n" +
                " owerAge INT,\n" +
                " owerIdcard STRING,\n" +
                " owerTel STRING,\n" +
                " owerHomeplace STRING,\n" +
                " owerWorkplace STRING,\n" +
                " remark STRING,\n" +
                " createTime TIMESTAMP,\n" +
                " operatorTime TIMESTAMP,\n" +
                " PRIMARY KEY (id) NOT ENFORCED\n" +
                ")  WITH (\n" +
                " 'connector' = 'jdbc',\n" +
                " 'url' = 'jdbc:mysql://master:3306/sca?useUnicode=true&characterEncoding=utf8',\n" +
                " 'table-name' = 'dim_base_ower',\n" +
                " 'username' = 'hive',\n" +
                " 'password' = 'hive'\n" +
                ")\n");
        //5.读取房间维度表
        tEnv.executeSql("CREATE TABLE dim_base_room(\n" +
                " id INT,\n" +
                " houseId INT,\n" +
                " roomCode STRING,\n" +
                " roomName STRING,\n" +
                " peopleNum INT,\n" +
                " remark STRING,\n" +
                " PRIMARY KEY (id) NOT ENFORCED\n" +
                ") WITH (\n" +
                " 'connector' = 'jdbc',\n" +
                " 'url' = 'jdbc:mysql://master:3306/sca?useUnicode=true&characterEncoding=utf8',\n" +
                " 'table-name' = 'dim_base_room',\n" +
                " 'username' = 'hive',\n" +
                " 'password' = 'hive'\n" +
                ")\n");
        //6.读取楼栋维度表
        tEnv.executeSql("CREATE TABLE dim_base_house(\n" +
                " id INT,\n" +
                " cid INT,\n" +
                " houseCode STRING,\n" +
                " houseName STRING,\n" +
                " buildStartTime TIMESTAMP,\n" +
                " buildEndTime TIMESTAMP,\n" +
                " houseArea FLOAT,\n" +
                " remark STRING,\n" +
                " PRIMARY KEY (id) NOT ENFORCED\n" +
                ")  WITH (\n" +
                " 'connector' = 'jdbc',\n" +
                " 'url' = 'jdbc:mysql://master:3306/sca?useUnicode=true&characterEncoding=utf8',\n" +
                " 'table-name' = 'dim_base_house',\n" +
                " 'username' = 'hive',\n" +
                " 'password' = 'hive'\n" +
                ")\n");
        //7.读取社区维度表
        tEnv.executeSql("CREATE TABLE dim_base_community(\n" +
                " id INT,\n" +
                " communityName STRING,\n" +
                " district_id INT,\n" +
                " city_id INT,\n" +
                " communityLocation STRING,\n" +
                " remark STRING,\n" +
                " PRIMARY KEY (id) NOT ENFORCED\n" +
                ") WITH (\n" +
                " 'connector' = 'jdbc',\n" +
                " 'url' = 'jdbc:mysql://master:3306/sca?useUnicode=true&characterEncoding=utf8',\n" +
                " 'table-name' = 'dim_base_community',\n" +
                " 'username' = 'hive',\n" +
                " 'password' = 'hive'\n" +
                ")\n");

        //8.事实表与维度表关联
        Table resultTable = tEnv.sqlQuery("select " +
                "egl.id, " +
                "egl.ownerId, " +
                "egl.openTime, " +
                "h.cid as cId, " +
                "egl.openMode " +
                "from ods_entrance_guard_log egl " +
                "left join dim_base_ower for system_time as of egl.proc_time as ol " +
                "on egl.ownerId=ol.id "+
                "left join dim_base_room for system_time as of egl.proc_time as  dbr "+
                "on ol.roomId=dbr.id "+
                "left join dim_base_house for system_time as of egl.proc_time as h "+
                "on dbr.houseId=h.id "+
                "left join dim_base_community for system_time as of egl.proc_time as c "+
                "on h.cid=c.id"
        );
        tEnv.createTemporaryView("resultTable",resultTable);
        //9.创建dwd_entrance_guard_log表
        tEnv.executeSql("CREATE TABLE dwd_entrance_guard_log (\n" +
                "id STRING,\n" +
                "ownerId INT,\n" +
                "openTime STRING,\n" +
                "cId INT,\n" +
                "openMode SMALLINT,\n" +
                "PRIMARY KEY (id) NOT ENFORCED\n" +
                ") WITH (\n" +
                "'connector' = 'upsert-kafka',\n" +
                "'topic' = 'dwd_entrance_guard_log',\n" +
                "'properties.bootstrap.servers' = 'master:9092',\n" +
                "'key.format' = 'json',\n" +
                "'value.format' = 'json'\n" +
                ")\n");
        //10.flink sql 将查询结果写入dwd_entrance_guard_log
        tEnv.executeSql("insert into dwd_entrance_guard_log select * from resultTable");
    }
}
